package DataStreamApi.Transformation算子.分流;

import domain.WaterSensor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Arrays;
import java.util.List;

/**
 * 1输入1输出
 */
public class Flink02_OUTPUT {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        List<WaterSensor> waterSensorList = Arrays.asList(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s2", 3L, 3),
                new WaterSensor("s3", 1L, 4),
                new WaterSensor("s1", 11L, 5),
                new WaterSensor("s1", 2L, 6));
        executionEnvironment.setParallelism(6);

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = executionEnvironment.fromCollection(waterSensorList);


        OutputTag<WaterSensor> s1Tag = new OutputTag<WaterSensor>("s1", Types.POJO(WaterSensor.class));
        OutputTag<WaterSensor> s2Tag = new OutputTag<WaterSensor>("s2", Types.POJO(WaterSensor.class));
        /**
         * new ProcessFunction<WaterSensor, WaterSensor>(),第一个参数指的是主流的输入类型，第二个参数指的是主流的输出类型
         */
        SingleOutputStreamOperator<WaterSensor> process = waterSensorDataStreamSource.process(new ProcessFunction<WaterSensor, WaterSensor>() {
            /**
             * WaterSensor value=输入
             * ProcessFunction<WaterSensor, WaterSensor>.Context ctx 上下文
             * Collector<WaterSensor> out 输出
             */
            @Override
            public void processElement(WaterSensor value, ProcessFunction<WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
                if (value.getId().equals("s1")) {

                    /**
                     * 第一个参数传入TAG对象
                     * 第二个参数传入输出流中的数据
                     */
                    ctx.output(s1Tag, value);
                } else if (value.getId().equals("s2")) {
                    ctx.output(s2Tag, value);
                } else {
                    out.collect(value);
                }

            }
        });

        /**
         * 这里打印的是主流数据，非S1S2
         */
        process.print("主流数据");


        /**
         * 打印侧输出流s1Tag
         */
        process.getSideOutput(s1Tag).print("s1Tag");


        /**
         * 打印侧输出流s2Tag
         */
        process.getSideOutput(s2Tag).print("s2Tag");
        executionEnvironment.execute();
    }
}
